package com.wz.jiangsu.controller.cep;

import cn.hutool.json.JSONUtil;
import com.wz.jiangsu.bean.LoginEvent;
import com.wz.jiangsu.bean.vo.show.ShowSecurityEventDataCommonWazuhDto;
import com.wz.jiangsu.util.JsonUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.SuccessCallback;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.concurrent.TimeUnit;

/**
 * @program: wz
 * @author: wangzhou
 * @create: 2024-02-29 17:19
 * @description:
 **/
@RestController
public class SendCEPMessageController {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;


    @PostMapping("/kafka/post/cep/sendMessage")
    public String sendCallbackOneMessageMoneyWazuhEngine(@RequestBody(required = false) ShowSecurityEventDataCommonWazuhDto dto, @RequestParam("topic") String topic) {


//            LoginEvent a1 = new LoginEvent("1", "192.168.10.1", false, 1000L);
//            LoginEvent a2 = new LoginEvent("2", "192.168.10.6", true, 2000L);
//            LoginEvent a3 = new LoginEvent("1", "192.168.10.1", false, 5000L);
//            LoginEvent a4 = new LoginEvent("2", "192.168.10.6", false, 5000L);
//            LoginEvent a5 = new LoginEvent("1", "192.168.10.1", false, 4000L);

        LoginEvent a1 = new LoginEvent("1", "192.168.10.1", false, 1000L);
        LoginEvent a2 = new LoginEvent("1", "192.168.10.6", false, 2000L);
        LoginEvent a3 = new LoginEvent("1", "192.168.10.1", false, 3000L);
        LoginEvent a4 = new LoginEvent("1", "192.168.10.6", false, 4000L);
        LoginEvent a5 = new LoginEvent("1", "192.168.10.1", false, 5000L);




        ArrayList<LoginEvent> list = new ArrayList<>();

        list.add(a1);
        list.add(a2);
        list.add(a3);
        list.add(a4);
        list.add(a5);

        list.forEach(n -> {
            String jsonStr = JSONUtil.toJsonStr(n);
            String oneRowJson = JsonUtil.replaceAllBlank(jsonStr);
            kafkaTemplate.send(topic, oneRowJson).addCallback(new SuccessCallback<SendResult<String, Object>>() {
                //成功的回调
                @Override
                public void onSuccess(SendResult<String, Object> success) {
                    // 消息发送到的topic
                    String topic = success.getRecordMetadata().topic();
                    // 消息发送到的分区
                    int partition = success.getRecordMetadata().partition();
                    // 消息在分区内的offset
                    long offset = success.getRecordMetadata().offset();

//                    System.out.println("发送消息成功-安全日志:" + topic + "-" + partition + "-" + offset);
                }
            }, new FailureCallback() {
                //失败的回调
                @Override
                public void onFailure(Throwable throwable) {
                    System.out.println("发送消息失败1:" + throwable.getMessage());
                }
            });
//            try {
//                TimeUnit.SECONDS.sleep(1);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
        });
        return "cep-data-5";
    }


    @PostMapping("/kafka/post/cep/sendMessage/always")
    public String sendCallbackOneMessageMoneyWazuhEngineData(@RequestBody(required = false) ShowSecurityEventDataCommonWazuhDto dto, @RequestParam("topic") String topic) {

        for (int i = 0; i < 1000; i++) {
            LoginEvent loginEvent = new LoginEvent("1", "192.168.10.1", false, i * 1000L);
            String jsonStr = JSONUtil.toJsonStr(loginEvent);
            String oneRowJson = JsonUtil.replaceAllBlank(jsonStr);
            kafkaTemplate.send(topic, oneRowJson).addCallback(new SuccessCallback<SendResult<String, Object>>() {
                //成功的回调
                @Override
                public void onSuccess(SendResult<String, Object> success) {
                    // 消息发送到的topic
                    String topic = success.getRecordMetadata().topic();
                    // 消息发送到的分区
                    int partition = success.getRecordMetadata().partition();
                    // 消息在分区内的offset
                    long offset = success.getRecordMetadata().offset();

//                    System.out.println("发送消息成功-安全日志:" + topic + "-" + partition + "-" + offset);
                }
            }, new FailureCallback() {
                //失败的回调
                @Override
                public void onFailure(Throwable throwable) {
                    System.out.println("发送消息失败1:" + throwable.getMessage());
                }
            });
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        return "cep-always";
    }

}
